package com.netty.disruptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.netty.model.SentBody;
import io.netty.channel.Channel;

import java.util.concurrent.Executors;

/**
 * @author zl
 * @version 1.0
 * @date 2021/10/8 13:35
 */
public class DisruptorUtils {

    private int ringBuffer = 1024*1024;
    private Disruptor disruptor ;
    /**
     * 创建disruptor
     */
    private void createDisruptor(){
        DisruptorConsumer[] disruptorConsumers = new DisruptorConsumer[10];
        for (int i = 0; i < disruptorConsumers.length; i++) {
            disruptorConsumers[i] = new DisruptorConsumer();
        }
        disruptor = new Disruptor(new EventFactory() {
            @Override
            public Object newInstance() {
                return new TranslatorDataWrapper();
            }
        },ringBuffer, Executors.defaultThreadFactory(), ProducerType.SINGLE,new BlockingWaitStrategy());
        //自定义异常
        disruptor.setDefaultExceptionHandler(new IntEventExceptionHandler());
        //不允许重复消费
        disruptor.handleEventsWithWorkerPool(disruptorConsumers);
        //允许重复消费
        //disruptor.handleEventsWith(disruptorConsumers);
        disruptor.start();
    }

    //发送消息到disruptor
    private void sendMessage(SentBody sentBody, Channel channel){
        RingBuffer<TranslatorDataWrapper> ringBuffer = disruptor.getRingBuffer();
        long sequence = ringBuffer.next();
        TranslatorDataWrapper translatorDataWrapper = ringBuffer.get(sequence);
        translatorDataWrapper.setChannel(channel);
        translatorDataWrapper.setSentBody(sentBody);
        ringBuffer.publish(sequence);
    }

}
//自定义异常
class IntEventExceptionHandler implements ExceptionHandler {
    public void handleEventException(Throwable ex, long sequence, Object event) {
        System.out.println(ex);
    }
    public void handleOnStartException(Throwable ex) {
        System.out.println(ex);
    }
    public void handleOnShutdownException(Throwable ex) {
        System.out.println(ex);
    }
}
